package com.chukun.multi.serialthreadconfinement.reusable;

import com.chukun.multi.twophase.AbstractTerminatableThread;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

/**
 *
 * @param <T> Serialiser 向WorkThread所提交的任务的类型
 * @param <V> 执行结果的类型
 */
public class TerminatableWorkerThread<T,V> extends AbstractTerminatableThread {


    private final BlockingQueue<Runnable> workQueue;

    private final TaskProcessor<T,V> taskProcessor;

    public TerminatableWorkerThread(BlockingQueue<Runnable> workQueue,TaskProcessor<T,V> taskProcessor) {
        this.workQueue = workQueue;
        this.taskProcessor = taskProcessor;
    }

    /**
     *提交任务到队列
     * @param task
     * @return
     * @throws InterruptedException
     */
    public Future<V> submit(final T task) throws InterruptedException{
        Callable<V> callable = new Callable<V>() {
            @Override
            public V call() throws Exception {
                return taskProcessor.doProcess(task);
            }
        };
        FutureTask<V> ft = new FutureTask<>(callable);
        workQueue.put(ft);
        terminationToken.reservations.incrementAndGet();
        return ft;
    }

    /**
     * 执行任务的逻辑
     * @throws Exception
     */
    @Override
    protected void doRun() throws Exception {
        Runnable take = workQueue.take();
        try{
            take.run();
        }finally {
            terminationToken.reservations.decrementAndGet();
        }
    }
}
